package com.hu.api

import org.apache.flink.api.common.accumulators.IntCounter
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2021/1/19 10:42
 * @Describe: 累加器只能针对会结束的flink程序才能得到累加器结果
 */
object IntCounterBounded {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.fromElements("a b c", "d d e f")
//    val inputStream = env.readTextFile("D:\\hujianjun\\T\\test\\hu\\huj.txt")
//    val inputStream = env.socketTextStream("localhost",6666) //得不得累加结果

    inputStream.flatMap(new RichFlatMapFunction[String, (String, Integer)] {
      private var inc: IntCounter = null


      override def open(parameters: Configuration): Unit = {
        inc = new IntCounter();
        getRuntimeContext().addAccumulator("word_nums", this.inc);
      }

      override def flatMap(value: String, out: Collector[(String, Integer)]): Unit = {
        val strings = value.split(" ")
        strings.foreach(s => {
          this.inc.add(1)
          out.collect((s, 1))
        })
      }
    }).print()

    println(env.execute("实现有界流累加器demo").getAccumulatorResult("word_nums"))
  }
}
